package com.yiruantong.common.websocket.utils;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.json.JSON;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import com.yiruantong.common.core.constant.Constants;
import com.yiruantong.common.core.domain.model.LoginUser;
import com.yiruantong.common.core.utils.StringUtils;
import com.yiruantong.common.json.utils.JsonUtils;
import com.yiruantong.common.redis.utils.RedisUtils;
import com.yiruantong.common.websocket.constant.WebSocketConstants;
import com.yiruantong.common.websocket.dto.WebSocketMessageDto;
import com.yiruantong.common.websocket.holder.WebSocketSessionHolder;
import org.springframework.web.socket.PongMessage;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;

import static com.yiruantong.common.websocket.constant.WebSocketConstants.LOGIN_USER_KEY;
import static com.yiruantong.common.websocket.constant.WebSocketConstants.WEB_SOCKET_TOPIC;

/**
 * 工具类
 *
 * @author zendwang
 */
@Slf4j
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class WebSocketUtils {

  /**
   * 发送消息
   *
   * @param sessionKey session主键 一般为用户id
   * @param message    消息文本
   */
  public static void sendMessage(String sessionKey, String message) {
    WebSocketSession session = WebSocketSessionHolder.getSessions(sessionKey);
    sendMessage(session, message);
  }

  /**
   * 订阅消息
   *
   * @param consumer 自定义处理
   */
  public static void subscribeMessage(Consumer<WebSocketMessageDto> consumer) {
    RedisUtils.subscribe(WEB_SOCKET_TOPIC, WebSocketMessageDto.class, consumer);
  }

  /**
   * 发布订阅的消息
   *
   * @param webSocketMessage 消息对象
   */
  public static void publishMessage(WebSocketMessageDto webSocketMessage) {
    List<String> unsentSessionKeys = new ArrayList<>();
    // 当前服务内session,直接发送消息
    for (String sessionKey : webSocketMessage.getSessionKeys()) {
      if (WebSocketSessionHolder.existSession(sessionKey)) {
        String message = webSocketMessage.getMessage();
        if (JsonUtils.isTypeJSON(message)) {
          JSON json = JsonUtils.parse(message);
          json.putByPath(Constants.WEB_SOCKET_SESSION_ID, sessionKey);
          message = JsonUtils.toJsonString(json);
        }
        WebSocketUtils.sendMessage(sessionKey, message);
        continue;
      }
      unsentSessionKeys.add(sessionKey);
    }
    // 不在当前服务内session,发布订阅消息
    if (CollUtil.isNotEmpty(unsentSessionKeys)) {
      WebSocketMessageDto broadcastMessage = new WebSocketMessageDto();
      broadcastMessage.setMessage(webSocketMessage.getMessage());
      broadcastMessage.setSessionKeys(unsentSessionKeys);
      RedisUtils.publish(WEB_SOCKET_TOPIC, broadcastMessage, consumer -> {
        log.info(" WebSocket发送主题订阅消息topic:{} session keys:{} message:{}",
          WEB_SOCKET_TOPIC, unsentSessionKeys, webSocketMessage.getMessage());
      });
    }
  }

  /**
   * 发布订阅的消息(群发)
   *
   * @param message 消息内容
   */
  public static void publishAll(String message) {
    WebSocketSessionHolder.getSessionsAll().forEach(key -> {
      WebSocketUtils.sendMessage(key, message);
    });
    WebSocketMessageDto broadcastMessage = new WebSocketMessageDto();
    broadcastMessage.setMessage(message);
    RedisUtils.publish(WEB_SOCKET_TOPIC, broadcastMessage, consumer -> {
      log.info(" WebSocket发送主题订阅消息topic:{} message:{}", WEB_SOCKET_TOPIC, message);
    });
  }

  public static void sendPongMessage(WebSocketSession session) {
    sendMessage(session, new PongMessage());
  }

  public static void sendMessage(WebSocketSession session, String message) {
    sendMessage(session, new TextMessage(message));
  }

  private static void sendMessage(WebSocketSession session, WebSocketMessage<?> message) {
    if (session == null || !session.isOpen()) {
      log.error("[send] session会话已经关闭");
    } else {
      try {
        // 获取当前会话中的用户
        LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
        session.sendMessage(message);
        if (!StringUtils.contains(message.getPayload().toString(), WebSocketConstants.PING)) {
          log.info("[send] sessionId: {},userId:{},userType:{},message:{}", session.getId(), loginUser.getUserId(), loginUser.getUserType(), message);
        }
      } catch (IOException e) {
        log.error("[send] session({}) 发送消息({}) 异常", session, message, e);
      }
    }
  }
}
